package com.company.poker.channel;

import akka.actor.ActorRef;
import akka.camel.Activation;
import akka.camel.CamelMessage;
import akka.camel.javaapi.UntypedConsumerActor;
import akka.dispatch.Mapper;
import com.company.poker.protocol.ByteMessage;
import org.apache.camel.model.ProcessorDefinition;
import org.apache.camel.model.RouteDefinition;
import org.jboss.netty.channel.UpstreamMessageEvent;

import java.util.Map;

public class MyEndpoint
    extends UntypedConsumerActor {
    private final ChannelContext context;
    private ActorRef table;


    public MyEndpoint(ChannelContext context) {
        this.context = context;
        table = context.getTable();
    }

    /**
     * Java API: Returns the [[akka.camel.Activation]] interface that can be used to wait on activation or
     * de-activation of Camel endpoints.
     * 
     * @return the Activation interface
     */
    @Override
    public Activation getActivation() {
        return super.getActivation();
    }

    @Override
    public String getEndpointUri() {
        return "akka://poker/user/MyEndpoint";
    }

    @Override
    public void onReceive(Object message) {
        if (message instanceof CamelMessage) {
            CamelMessage msg = (CamelMessage) message;

            Map<String,Object> headers = msg.getHeaders();
            UpstreamMessageEvent camelNettyMessageEvent = (UpstreamMessageEvent) headers.get("CamelNettyMessageEvent");
            String id = (String) headers.get("breadcrumbId");
            context.getSessionCash().put(id,camelNettyMessageEvent.getChannel());

            System.out.println("!!!!!!!!!!" + msg.body());
            System.out.println("!!!!!!!!!!" + getSender());
            table.tell(msg.body(), getSelf());
        } else {
            ByteMessage msg = (ByteMessage) message;
            getSender().tell(msg);
        }
    }

    @Override
    public void preStart() {
        super.preStart();
        System.out.println("@@@@@@@@@@ MyEndpoint - preStart");
    }

    /**
     * Java API: Returns the [[akka.dispatch.Mapper]] function that will be used as a route definition handler
     * for creating custom route to this consumer. By default it returns an identity function, override this method to
     * return a custom route definition handler. The [[akka.dispatch.Mapper]] is not allowed to close over 'this', meaning it is
     * not allowed to refer to the actor instance itself, since that can easily cause concurrent shared state issues.
     */
    @Override
    public Mapper<RouteDefinition, ProcessorDefinition<?>> getRouteDefinitionHandler() {
        return super.getRouteDefinitionHandler();
    }
}
